package com.baishancloud.log.common.connector.kafka

import com.baishancloud.log.common.util.ParamUtil.paramName
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer

import java.io.Serializable
import java.util.{Objects, Properties}
import scala.beans.BeanProperty
import scala.collection.JavaConverters.seqAsJavaListConverter
import scala.language.postfixOps

/**
 * 提供kafka source连接器的获取功能
 *
 * @author ziqiang.wang
 * @date 2021/10/18 17:23
 */
class KafkaSourceConnector extends Serializable {
  /**
   * 必选<br>
   * 连接kafka的url，示例：broker1:9092,broker2:9092
   */
  @BeanProperty var bootstrapServers: String = ""
  /**
   * 必选<br>
   * 主题名称，多个名称，使用英文逗号分隔，中间不要有空格
   */
  @BeanProperty var topics: String = ""
  /**
   * 必选<br>
   * 组id
   */
  @BeanProperty var groupId: String = ""
  /**
   * 第一次消费kafka时的offset位置，默认从最开始位置消费<br>
   * earliest，表示最早位置；latest表示最新位置。
   */
  @BeanProperty var autoOffsetReset: OffsetsInitializer = OffsetsInitializer.latest()
  /**
   * 是否自动提交offset，默认值为true<br>
   * 注：如果开启了checkpoint，则自动提交offset会失效，任务将根据checkpoint来提交offset
   */
  @BeanProperty var enableAutoCommit: Boolean = true
  /**
   * 自动提交offset的情况下，自动提交的间隔，默认5秒，单位：秒
   */
  @BeanProperty var autoCommitIntervalMs: Long = 5 * 1000
  /**
   * 自动发现新主题和分区的时间间隔，默认为一小时，单位：分钟
   */
  @BeanProperty var partitionDiscoveryIntervalMs: Long = 60 * 60 * 1000
  /**
   * 用来保存其他的一些设置
   */
  @BeanProperty val properties: Properties = new Properties()


  /**
   * 完成设置对象内容的最后一些操作
   */
  def build(): KafkaSourceConnector = {
    properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit.toString)
    properties.setProperty(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "120000")
    if (enableAutoCommit) {
      properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs.toString)
    }
    this
  }


  /**
   * 只返回value数据
   *
   * @return KafkaSource对象
   */
  def sourceOnlyValue(): KafkaSource[String] = {
    KafkaSource.builder()
      .setBootstrapServers(bootstrapServers)
      .setTopics(topics.split(",").toList.asJava)
      .setGroupId(groupId)
      .setStartingOffsets(autoOffsetReset)
      .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(classOf[StringDeserializer]))
      .setProperties(properties)
      .build()
  }

  /**
   * 返回更多kafka消息信息<br>
   * 可获取数据，具体参考 [[JSONKeyValueDeserializationSchema]]。
   *
   * @param includeMetaData 是否包含元数据，默认不包含
   * @return KafkaSource对象
   */
  def sourceObjectNode(includeMetaData: Boolean = false): KafkaSource[ObjectNode] = {
    KafkaSource.builder()
      .setBootstrapServers(bootstrapServers)
      .setTopics(topics.split(",").toList.asJava)
      .setGroupId(groupId)
      .setStartingOffsets(autoOffsetReset)
      .setDeserializer(KafkaRecordDeserializationSchema.of(new JSONKeyValueDeserializationSchema(includeMetaData)))
      .setProperties(properties)
      .build()
  }


}


/**
 * 提供kafka连接器的source和sink的获取函数，简化flink流应用开发中获取kafka连接器的重复代码开发
 *
 * @author ziqiang.wang
 * @date 2021/10/18 16:28
 */
object KafkaSourceConnector {

  /**
   * 构建方法，返回消费者连接器参数对象，后面可以通过这个对象来获取需要用到的属性<br>
   * 主参数输入值示例：--source.bootstrapServers或者是--source1.bootstrapServers<br>
   * 具体参数，可参考包对象
   *
   * @param parameterTool 参数工具类，包含了主方法接收到的参数
   * @param number        要获取那个数字对应的数据源
   */
  def builder(parameterTool: ParameterTool, number: String = ""): KafkaSourceConnector = {
    val connector: KafkaSourceConnector = new KafkaSourceConnector

    if (parameterTool.has(paramName(sourceBootstrapServersP, number))) {
      connector.setBootstrapServers(parameterTool.get(paramName(sourceBootstrapServersP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sourceBootstrapServersP, number)} 参数，可能是在source后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sourceTopicsP, number))) {
      connector.setTopics(parameterTool.get(paramName(sourceTopicsP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sourceTopicsP, number)} 参数，可能是在source后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sourceGroupIdP, number))) {
      connector.setGroupId(parameterTool.get(paramName(sourceGroupIdP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sourceGroupIdP, number)} 参数，可能是在source后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sourceAutoOffsetResetP, number))) {
      if (Objects.equals(earliest, parameterTool.get(paramName(sourceAutoOffsetResetP, number)))) {
        connector.setAutoOffsetReset(OffsetsInitializer.earliest())
      } else if (Objects.equals(latest, parameterTool.get(paramName(sourceAutoOffsetResetP, number)))) {
        connector.setAutoOffsetReset(OffsetsInitializer.latest())
      }
    }

    if (parameterTool.has(paramName(sourceEnableAutoCommitP, number))) {
      connector.setEnableAutoCommit(parameterTool.getBoolean(paramName(sourceEnableAutoCommitP, number)))
    }

    if (parameterTool.has(paramName(sourceAutoCommitIntervalMsP, number))) {
      connector.setAutoCommitIntervalMs(parameterTool.getLong(paramName(sourceAutoCommitIntervalMsP, number) * 1000))
    }

    if (parameterTool.has(paramName(sourcePartitionDiscoveryIntervalMsP, number))) {
      connector.setPartitionDiscoveryIntervalMs(parameterTool.getLong(sourcePartitionDiscoveryIntervalMsP) * 60 * 1000)
    }

    //其他参数
    parameterTool.toMap.entrySet().forEach(entry => {
      val key: String = entry.getKey
      if (key.startsWith("source" + number) && key.contains(".properties.")) {
        connector.properties.put(key.substring(key.indexOf(46, key.indexOf(".") + 1) + 1), entry.getValue)
      }
    })

    connector
  }


}